package kafka_learn

import (
	"context"
	"fmt"
	"github.com/IBM/sarama"
	"sync"
	"time"
)

type ConsumerGroupHandler struct{}

func (ConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (ConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h ConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		time.Sleep(time.Second)
		fmt.Printf("Message topic:%q partition:%d offset:%d, value:%s\n", msg.Topic, msg.Partition, msg.Offset, msg.Value)
		sess.MarkMessage(msg, "")
	}
	return nil
}

func ConsumerGroup() {
	broker := []string{"127.0.0.1:9092", "127.0.0.1:9101"}
	config := sarama.NewConfig()
	config.Consumer.Offsets.Initial = sarama.OffsetOldest
	config.Consumer.Return.Errors = true
	groupName := "my-group-4"
	group, err := sarama.NewConsumerGroup(broker, groupName, config)
	if err != nil {
		panic(err)
	}
	defer group.Close()

	go func() {
		for err := range group.Errors() {
			fmt.Println(err)
		}
	}()
	ctx := context.Background()
	topic1 := "test01"
	for i := 0; i < 3; i++ {
		go func(i int) {
			fmt.Println("consumer", i)
			err = group.Consume(ctx, []string{topic1}, &ConsumerGroupHandler{})
			if err != nil {
				panic(err)
			}
		}(i)
	}
	quit := make(chan struct{})
	<-quit
}

func Consumer() {
	broker := []string{"127.0.0.1:9092", "127.0.0.1:9101"}
	config := sarama.NewConfig()
	config.Consumer.Offsets.Initial = sarama.OffsetOldest
	config.Consumer.Return.Errors = true

	consumer, err := sarama.NewConsumer(broker, config)
	if err != nil {
		panic(err)
	}
	defer consumer.Close()
	topic := "test02"
	partList, err := consumer.Partitions(topic)
	if err != nil {
		panic(err)
	}
	var wg sync.WaitGroup
	for _, part := range partList {
		wg.Add(1)
		pc, err := consumer.ConsumePartition(topic, part, 2)
		if err != nil {
			panic(err)
		}
		defer pc.AsyncClose()
		go func(pc sarama.PartitionConsumer) {
			defer wg.Done()
			for msg := range pc.Messages() {
				fmt.Printf("Message topic:%q partition:%d offset:%d, value:%s\n", msg.Topic, msg.Partition, msg.Offset, msg.Value)
			}
		}(pc)
	}
	wg.Wait()
}
